<?php
namespace app\data\command;
use Psr\Log\NullLogger;
use Stomp\Client;
use Stomp\Exception\StompException;
use Stomp\Network\Observer\Exception\HeartbeatException;
use Stomp\Network\Observer\ServerAliveObserver;
use Stomp\StatefulStomp;
use think\admin\Command;
use think\admin\Exception;
use think\admin\service\ProcessService;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;
use app\data\service\IotMsgService;
/**
 * 阿里云  CmdAliyunStomp
 *
sudo -u www /www/server/php/74/bin/php /www/wwwroot/tatatest/think xapi:CmdAliyunStomp start 开启
sudo -u www /www/server/php/74/bin/php /www/wwwroot/tatatest/think xapi:CmdAliyunStomp start -d 守卫进程
sudo -u www /www/server/php/74/bin/php /www/wwwroot/tatatest/think xapi:CmdAliyunStomp stop 关闭
sudo -u www /www/server/php/74/bin/php /www/wwwroot/tatatest/think xapi:CmdAliyunStomp status 状态
sudo -u www /www/server/php/74/bin/php /www/wwwroot/tatatest/think xapi:CmdAliyunStomp listen 监听
php think xapi:CmdAliyunStomp start -d 守卫进程
php think xapi:CmdAliyunStomp stop  关闭
php think xapi:CmdAliyunStomp status 状态
php think xapi:CmdAliyunStomp listen 监听
 */
class CmdAliyunStomp extends Command
{
    /**
     * php think  xapi:CmdAliyunStomp start -d
     * Class CmdAliyunStomp
     * @package app\data\command
     */

    const QUEUE_LISTEN = 'xapi:CmdAliyunStomp listen';

    protected $name = 'xapi:CmdAliyunStomp';

    //不能重复创建;
    protected function configure()
    {
        $this->setName($this->name);
        $this->addOption('daemon', 'd', Option::VALUE_NONE, 'The queue listen in daemon mode');
        $this->addArgument('action', Argument::OPTIONAL, 'stop|start|status|listen', 'listen');
        $this->addArgument('code', Argument::OPTIONAL, 'Taskcode');
        $this->addArgument('spts', Argument::OPTIONAL, 'Separator');
        $this->setDescription('xapi:CmdTest listen for my');

    }

    /**
     * 业务指令执行
     * @param Input $input
     * @param Output $output
     * @return void
     * @throws Exception
     */
    protected function execute(Input $input, Output $output)
    {

        $action = $input->hasOption('daemon') ? 'start' : $input->getArgument('action');
        if (method_exists($this, $method = "{$action}Action"))
            return $this->$method();

        $this->output->error("># Wrong operation, Allow stop|start|status|listen");
    }


    /**
     * 停止所有任务
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\DbException
     * @throws \think\db\exception\ModelNotFoundException
     */
    protected function stopAction()
    {
        if (count($result = $this->process->thinkQuery($this->name)) < 1) {
            $this->output->writeln("># There are no task processes to stop");
        } else foreach ($result as $item) {
            $this->process->close(intval($item['pid']));
            $this->output->writeln("># Successfully sent end signal to process {$item['pid']}");
        }
    }

    /**
     * 启动后台任务
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\DbException
     * @throws \think\db\exception\ModelNotFoundException
     */
    protected function startAction()
    {
        // SystemQueue::mk()->count();
        $this->output->comment(">$ {$this->process->think(static::QUEUE_LISTEN)}");
        if (count($result = $this->process->thinkQuery(static::QUEUE_LISTEN)) > 0) {
            $this->output->writeln("># " . static::QUEUE_LISTEN . " already exist for pid {$result[0]['pid']}");
        } else {
            $this->process->thinkExec(static::QUEUE_LISTEN, 1000);
            if (count($result = $this->process->thinkQuery(static::QUEUE_LISTEN)) > 0) {

                $this->output->writeln("># " . static::QUEUE_LISTEN . " started successfully for pid {$result[0]['pid']}");

            } else {
                $this->output->writeln("># " . static::QUEUE_LISTEN . " failed to start");

            }
        }
    }

    /**
     * 查询兼听状态
     * @throws \think\db\exception\DataNotFoundException
     * @throws \think\db\exception\DbException
     * @throws \think\db\exception\ModelNotFoundException
     */
    protected function statusAction()
    {
        if (count($result = $this->process->thinkQuery(static::QUEUE_LISTEN)) > 0) {
            $this->output->writeln("Listening for main process {$result[0]['pid']} running");
        } else {
            $this->output->writeln("The Listening main process is not running");
        }
    }

    /**
     * 立即监听任务
     */
    protected function listenAction()
    {
        set_time_limit(0);
        ignore_user_abort(true);
        $this->app->db->setLog(new NullLogger());
        $this->output->writeln("\tYou can exit with <info>`CTRL-C`</info>");
        $this->output->writeln('=============== LISTENING ===============');

        $this->begin();

    }

    protected function begin()
    {
        //参数说明，请参见AMQP客户端接入说明文档。
        $accessKey = config('iot.accessKeyId');
        $accessSecret =  config('iot.accessKeySecret');
        $consumerGroupId = "DEFAULT_GROUP";
        $clientId = "your_milk";
        $iotInstanceId = config('iot.IotInstanceId');//iotInstanceId：实例ID。
        $timeStamp = round(microtime(true) * 1000);
        $signMethod = "hmacsha1";//签名方法
        //userName组装方法，请参见AMQP客户端接入说明文档。
        //若使用二进制传输，则userName需要添加encode=base64参数，服务端会将消息体base64编码后再推送。具体添加方法请参见下一章节“二进制消息体说明”。
        $userName = $clientId . "|authMode=aksign"
            . ",signMethod=" . $signMethod
            . ",timestamp=" . $timeStamp
            . ",authId=" . $accessKey
            . ",iotInstanceId=" . $iotInstanceId
            . ",consumerGroupId=" . $consumerGroupId
            . "|";
        $signContent = "authId=" . $accessKey . "&timestamp=" . $timeStamp;
        //计算签名，password组装方法，请参见AMQP客户端接入说明文档。
        $password = base64_encode(hash_hmac("sha1", $signContent, $accessSecret, $raw_output = TRUE));
        $client = new Client( config('iot.amqp_url'));
        $sslContext = ['ssl' => ['verify_peer' => true, 'verify_peer_name' => false],];
        $client->getConnection()->setContext($sslContext);
        //服务端心跳监听。
        $observer = new ServerAliveObserver();
        $client->getConnection()->getObservers()->addObserver($observer);
        //心跳设置，需要云端每30s发送一次心跳包。
        $client->setHeartbeat(0, 3000);
        $client->setLogin($userName, $password);
        try {
            $client->connect();
        } catch (StompException $e) {
            log_result("failed to connect to server, msg:" . $e->getMessage(),'stomp_error.txt');
            echo "failed to connect to server, msg:" . $e->getMessage(), PHP_EOL;
        }
        //无异常时继续执行。
        $stomp = new StatefulStomp($client);
        $stomp->subscribe('/topic/#');
        $this->output->writeln('connect success .');
        while (true) {
            $this->output->writeln('ok');
            try {
                // 检查连接状态
                if (!$client->isConnected()) {
                    echo "connection not exists, will reconnect after 10s.", PHP_EOL;
                    sleep(10);
                    $client->connect();
                    $stomp->subscribe('/topic/#');
                    echo "connect success.", PHP_EOL;
                }
                $msg = $stomp->read();
                echo date('Y-m-d H:i:s');
                echo $msg;
                if (!empty($msg)) {
                    // 处理消息业务逻辑。
                    $msgService = new IotMsgService();
                    $msgService->handle($msg->getCommand(),$msg->getHeaders(),$msg->getBody());
                }

            } catch (HeartbeatException $e) {
                echo 'The server failed to send us heartbeats within the defined interval.', PHP_EOL;
                log_result('The server failed to send us heartbeats within the defined interval.','stomp_error.txt');
                $stomp->getClient()->disconnect();
            } catch (Exception $e) {
                echo 'process message occurs error: ' . $e->getMessage(), PHP_EOL;
                log_result('process message occurs error: ' . $e->getMessage(),'stomp_error.txt');
                $stomp->getClient()->disconnect();
            }
        }
    }
}